package org.baojie.raft.connector.server;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import org.baojie.raft.box.Box;
import org.baojie.raft.connector.server.initializer.ServerChannelInit;
import org.baojie.raft.util.current.RaftFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;

public class RaftNettyServer {

    private static final Logger log = LoggerFactory.getLogger(RaftNettyServer.class);
    private final ReentrantLock mainLock = new ReentrantLock();
    private volatile boolean start = false;
    private final boolean epollAvailable;
    private final int workNums;
    private final Box box;

    private final EventLoopGroup boss;
    private final EventLoopGroup worker;

    public RaftNettyServer(int workNums, Box box) {
        this.box = box;
        this.workNums = workNums;
        this.epollAvailable = Epoll.isAvailable();
        this.boss = eventLoop();
        this.worker = eventLoop();
    }

    public final boolean start() {
        final ReentrantLock lock = mainLock;
        lock.lock();
        try {
            if (!start) {
                ChannelFuture future = startByLock();
                if (null == future) {
                    start = false;
                } else {
                    start = true;
                }
                return start;
            } else {
                return true;
            }
        } finally {
            lock.unlock();
        }
    }

    private final ChannelFuture startByLock() {
        ServerBootstrap bootstrap = new ServerBootstrap();
        int port = box.name().getAddress().getPort();
        boolean suc = group(boss, worker, bootstrap);
        if (suc) {
            channel(bootstrap);
            handler(bootstrap, port);
            ChannelFuture future = asyncBind(bootstrap, port);
            if (null == future) {
                shutdownGracefully(boss, worker);
                return null;
            } else {
                final AtomicBoolean test = new AtomicBoolean(true);
                future.addListener(new ChannelFutureListener() {
                    public void operationComplete(ChannelFuture future) {
                        if (!future.isSuccess()) {
                            test.compareAndSet(true, false);
                            Throwable err = future.cause();
                            log.error("bind occur error", err);
                            shutdownGracefully(boss, worker);
                        }
                    }
                });
                if (test.get()) {
                    return future;
                } else {
                    return null;
                }
            }
        } else {
            shutdownGracefully(boss, worker);
            return null;
        }
    }

    private final ChannelFuture asyncBind(ServerBootstrap bootstrap, int port) {
        ChannelFuture future = null;
        try {
            future = bootstrap.bind(port);
        } catch (Throwable t) {

        }
        return future;
    }

    private final void shutdownGracefully(EventLoopGroup boss, EventLoopGroup worker) {
        try {
            if (null != boss) {
                boss.shutdownGracefully();
            }
        } finally {
            if (null != worker) {
                worker.shutdownGracefully();
            }
        }
    }

    private final EventLoopGroup eventLoop() {
        if (epollAvailable) {
            RaftFactory factory = RaftFactory.create("epoll raft netty server worker", false);
            return new EpollEventLoopGroup(workNums, factory);
        } else {
            RaftFactory factory = RaftFactory.create("nio raft netty server worker", false);
            return new NioEventLoopGroup(workNums, factory);
        }
    }

    private final boolean group(EventLoopGroup boss, EventLoopGroup worker, ServerBootstrap bootstrap) {
        boolean suc = false;
        try {
            ServerBootstrap copy = bootstrap.group(boss, worker);
            if (null != copy) {
                suc = true;
            } else {
                suc = false;
            }
        } catch (Throwable t) {
            suc = false;
        }
        return suc;
    }

    private final void channel(ServerBootstrap bootstrap) {
        if (epollAvailable) {
            bootstrap.channel(EpollServerSocketChannel.class);
        } else {
            bootstrap.channel(NioServerSocketChannel.class);
        }
    }

    private final void handler(ServerBootstrap bootstrap, int port) {
        bootstrap.handler(new LoggingHandler(LogLevel.DEBUG));
        bootstrap.localAddress(new InetSocketAddress(port));
        bootstrap.childHandler(new ServerChannelInit(box));
        bootstrap.option(ChannelOption.SO_BACKLOG, 1024);
        bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
    }

    public void stop() {
        final ReentrantLock lock = mainLock;
        lock.lock();
        try {
            if (start) {
                shutdownGracefully(worker, boss);
                start = false;
            }
        } finally {
            lock.unlock();
        }
    }

}
